package com.example.springboot.controller.guava;

import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * sse实现消息推送
 *SE与WebSocket作用相似，都可以建立服务端与浏览器之间的通信，实现服务端向客户端推送消息，但还是有些许不同：
 *
 * SSE 是基于HTTP协议的，它们不需要特殊的协议或服务器实现即可工作；WebSocket需单独服务器来处理协议。
 * SSE 单向通信，只能由服务端向客户端单向通信；webSocket全双工通信，即通信的双方可以同时发送和接受信息。
 * SSE 实现简单开发成本低，无需引入其他组件；WebSocket传输数据需做二次解析，开发门槛高一些。
 * SSE 默认支持断线重连；WebSocket则需要自己实现。
 * SSE 只能传送文本消息，二进制数据需要经过编码后传送；WebSocket默认支持传送二进制数据。
 *
 * @Author:ChenZhangKun
 * @Date: 2022/8/23 13:03
 */
@Slf4j
@RestController
public class SsePushingController {
    private static final Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();
    private static AtomicInteger count = new AtomicInteger(0);

    public static SseEmitter connect(String userId) {
        try {
            // 设置超时时间，0表示不过期。默认30秒
            SseEmitter sseEmitter = new SseEmitter(0L);
            // 注册回调
            sseEmitter.onCompletion(completionCallBack(userId));
            sseEmitter.onError(errorCallBack(userId));
            sseEmitter.onTimeout(timeoutCallBack(userId));
            sseEmitterMap.put(userId, sseEmitter);
            count.getAndIncrement();
            return sseEmitter;
        } catch (Exception e) {
            log.info("创建新的sse连接异常，当前用户：{}", userId);
        }
        return null;
    }

    private static Runnable timeoutCallBack(String userId) {
        return new Runnable() {
            @Override
            public void run() {
                log.info(userId + "超时");
            }
        };
    }

    private static Consumer<Throwable> errorCallBack(String userId) {
        return new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) {
                log.info(userId);
            }
        };
    }

    private static Runnable completionCallBack(String userId) {
        return new Runnable() {
            @Override
            public void run() {
                log.info(userId);
            }
        };
    }

    /**
     * 给指定用户发送消息
     *
     * @date: 2022/7/12 14:51
     * @auther: 公众号：程序员小富
     */
    public static void sendMessage(String userId, String message) {

        if (sseEmitterMap.containsKey(userId)) {
            try {
                sseEmitterMap.get(userId).send(message);
            } catch (IOException e) {
                log.error("用户[{}]推送异常:{}", userId, e.getMessage());
                removeUser(userId);
            }
        }

    }

    private static void removeUser(String userId) {
        sseEmitterMap.remove(userId);
    }
}
